#include "StreamServer.h"
#include "log_api.h"
#include "bigendian.h"

StreamServer::StreamServer(int count) : TcpSocketServer(count)
{
    for(int i = 0;i < MAX_CLIENT_NUM;i++)
    {
        m_FreeQueue.EnQueue(&m_ClientList[i].elem);
    }
    m_msg = (char*)malloc(MAX_STREAM_SIZE);
    m_msglen = 0;

    m_v4l2Capture.Initialize(/*m_pServer*/);
}

StreamServer::~StreamServer()
{
   OSQueueElem* pElem = NULL;
   do{
        pElem = m_WorkingQueue.DeQueue();
   }while(pElem);
   do{
        pElem = m_FreeQueue.DeQueue();
   }while(pElem);
   free(m_msg);
}

tClientInfo* StreamServer::Malloc()
{
    OSQueueElem* pElem = m_FreeQueue.DeQueue();
    if(pElem)
    {
        m_WorkingQueue.EnQueue(pElem);
        return (tClientInfo*)pElem->GetEnclosingObject();
    }
    return NULL;
}
void StreamServer::Cleanup()
{

}
void StreamServer::Free(tClientInfo* pClientInfo)
{
    if(pClientInfo)
    {
        m_WorkingQueue.Remove(&pClientInfo->elem);
        m_FreeQueue.EnQueue(&pClientInfo->elem);
    }
}
void StreamServer::Parse(int msgtype)
{
    switch(msgtype)
    {
        case M_SendHeartBeat:
            LOG_OUTPUT("Recv HeartBeat from Client\n");
            break;
        default:
            break;
    }
}
void StreamServer::ReadEvent(ConnNode *conn)
{
    /*
    int readable = 0;
    cfg_msg *pmsg = NULL;
    char pbuf[MIN_MSG_LEN] = { 0 };
    int length = conn->GetReadBufferLen();
    if (length < MIN_MSG_LEN)
    {
        return;
    }
    else
    {
        while (length >= MIN_MSG_LEN)
        {
            //先拷贝MIN_MSG_LEN字节数据
            readable = conn->CopyReadBuffer(pbuf, MSG_HDR_LEN);
            assert(readable == MSG_HDR_LEN);
            pmsg = (cfg_msg*)&pbuf[0];
            if (!pmsg->Isvalid())
            {
                return;
            }
            else if (length < pmsg->len)
            {
                return;
            }

            conn->DeleteFromReadBuffer(MSG_HDR_LEN);


            m_msglen = pmsg->len - MSG_HDR_LEN;
            //m_msgData.body = (char*)malloc(m_msgData.len);
            //m_msgBody = (char*)malloc(m_msgData.len - MSG_HDR_LEN);
            //memset(m_msgBody, 0, m_msgData.len - MSG_HDR_LEN);
            readable = conn->GetReadBuffer(m_msg, m_msglen);
            assert(readable == m_msglen);
            Parse(pmsg->type);
            length -= pmsg->len;
        }
    }
    */
    conn->MoveBufferData();
}

void StreamServer::WriteEvent(ConnNode *conn)
{
}

void StreamServer::ConnectionEvent(ConnNode *conn)
{
	StreamServer *me = (StreamServer*)conn->GetThread()->tcpConnect;
	printf("new connection: %d\n", conn->GetFd());
    tClientInfo* pClientInfo = Malloc();
    if(pClientInfo)
    {
        pClientInfo->pClient = conn;
    }
}

void StreamServer::CloseEvent(ConnNode *conn, short events)
{
    printf("CloseEvent Start: %d\n", conn->GetFd());
	StreamServer *me = (StreamServer*)conn->GetThread()->tcpConnect;
	for(OSQueueIter iter(&m_WorkingQueue);!iter.IsDone();)
    {
        OSQueueElem* pElem = iter.GetCurrent();
        iter.Next();
        if(NULL == pElem)
            continue;
        tClientInfo* pClientInfo = (tClientInfo*)pElem->GetEnclosingObject();
        if(NULL == pClientInfo)
            continue;

        if(pClientInfo->pClient == conn)
        {
            printf("Free Client\n");
            Free(pClientInfo);
            break;
        }
    }
	printf("connection closed: %d\n", conn->GetFd());
}

void StreamServer::TimeOutCb(int id, short events, void *data)
{

    struct StreamPackageHeader packageHeader;
    unsigned int packetNum = 0;
    unsigned int lastPacketSize = 0;
    unsigned char* framebuffer;
    unsigned int currentPacketIndex = 0;
    int framesize = 0;
    StreamServer *me = (StreamServer*)data;
    OSMutex *pMutex = me->m_v4l2Capture.GetMutex();
    OSCond *pCond = me->m_v4l2Capture.GetCond();



    //fprintf(stderr,"StreamServer:mutex=%p\n",pMutex);
    //fprintf(stderr,"StreamServer:cond=%p\n",pCond);
    OSMutexLocker locker(pMutex);
    //mutex.Lock();
    pCond->Wait(pMutex);
    //LOG_OUTPUT("Signal");
    framebuffer = me->m_v4l2Capture.GetFrameBuffer();
    framesize = me->m_v4l2Capture.GetFrameSize();
    //printf("framesize=%d\n",framesize);

    packetNum = framesize / MAX_EVER_SEND_LEN;
    lastPacketSize = framesize % MAX_EVER_SEND_LEN;
    if(lastPacketSize != 0)
    {
        packetNum++;
    }

    packageHeader.uPackageHeaderSize = sizeof(struct StreamPackageHeader);
    packageHeader.uTotalDataSize = framesize;
    packageHeader.uDataPackageNum = packetNum;
    memset(me->m_msg,0,MAX_STREAM_SIZE);

    //OSMutexLocker locker1(&me->m_mutex);
    while(currentPacketIndex < packetNum)
    {
        if(currentPacketIndex < (packetNum - 1))
        {
            packageHeader.uCurrentTransPackageSize = sizeof(struct StreamPackageHeader) + MAX_EVER_SEND_LEN;
            packageHeader.uDataPackageCurrIndex = currentPacketIndex + 1;
            packageHeader.uDataPackageOffset = currentPacketIndex * MAX_EVER_SEND_LEN;
            memcpy(me->m_msg,&packageHeader,sizeof(struct StreamPackageHeader));
            memcpy(me->m_msg+sizeof(struct StreamPackageHeader),framebuffer+packageHeader.uDataPackageOffset,MAX_EVER_SEND_LEN);
            //printf("packageHeader.uDataPackageCurrIndex=%d,size=%d\n",packageHeader.uDataPackageCurrIndex,packageHeader.uCurrentTransPackageSize);
            while(OS_NotEnoughMem == me->SendToClient(me->m_msg,packageHeader.uCurrentTransPackageSize))
                // 接收方收取数据太慢
                OSThread::Sleep(10);
            currentPacketIndex++;
        }
        else
        {
            packageHeader.uCurrentTransPackageSize = sizeof(struct StreamPackageHeader) + (framesize - currentPacketIndex * MAX_EVER_SEND_LEN );
            packageHeader.uDataPackageCurrIndex = currentPacketIndex + 1;
            packageHeader.uDataPackageOffset = currentPacketIndex * MAX_EVER_SEND_LEN;
            memcpy(me->m_msg,&packageHeader,sizeof(struct StreamPackageHeader));
            memcpy(me->m_msg+sizeof(struct StreamPackageHeader),framebuffer+packageHeader.uDataPackageOffset,framesize - currentPacketIndex * MAX_EVER_SEND_LEN);
            //printf("last totalSize=%d,framesize - currentPacketIndex * MAX_EVER_SEND_LEN=%d\n",packageHeader.uTotalDataSize,framesize - currentPacketIndex * MAX_EVER_SEND_LEN);
            //printf("last packageHeader.uDataPackageCurrIndex=%d,size=%d\n",
            //    packageHeader.uDataPackageCurrIndex, packageHeader.uCurrentTransPackageSize);
            while(OS_NotEnoughMem == me->SendToClient(me->m_msg,packageHeader.uCurrentTransPackageSize))
                // 接收方收取数据太慢
                OSThread::Sleep(10);
            currentPacketIndex++;
        }
    }

    // 接收方收取数据太慢
     OSThread::Sleep(10);
    //unsigned char* framebuffer;
    //int framesize = 0;
	//StreamServer *me = (StreamServer*)data;
    //OSMutex *pMutex = me->m_v4l2Capture.GetMutex();
    //OSCond *pCond = me->m_v4l2Capture.GetCond();

    //fprintf(stderr,"StreamServer:mutex=%p\n",pMutex);
    //fprintf(stderr,"StreamServer:cond=%p\n",pCond);
    //OSMutexLocker locker(pMutex);
    //mutex.Lock();
    //pCond->Wait(pMutex);
    //LOG_OUTPUT("Signal");
    //framebuffer = me->m_v4l2Capture.GetFrameBuffer();
    //framesize = me->m_v4l2Capture.GetFrameSize();
    //printf("framesize=%d\n",framesize);
    //if(framebuffer != NULL && framesize > 0)
    //    me->SendStreamToClient((char*)framebuffer,framesize);

    //SplitPackageAndSend(me);
    //char buf[1024]={0};
    //memcpy(buf,"hello\nworld!",11);
    //me->SendToClient(M_SendParameter,buf,11);
	//for (int i = 0; i<me->vec.size(); i++)
	//	me->vec[i]->AddToWriteBuffer((char*)&msg, sizeof(msg));
}
void StreamServer::HeartBeatCb(int id, int short events, void *data)
{
    StreamServer *me = (StreamServer*)data;
    LOG_OUTPUT("Send HeartBeat to Client\n");
    //me->SendToClient(M_SendHeartBeat,NULL,0);
}
OS_Error StreamServer::SendToClient(msg_type t, char* pbody, int len)
{
    /*
    cfg_msg msg;
    msg.type = t;
    msg.len = MSG_HDR_LEN + len;

    if(len > MAX_MSG_LEN)
    {
        LOG_OUTPUT("Error:invalid len,T = %d,len:%d,\n",msg.typestr(),len);
        return OS_InvalidArgument;
    }

    if(len > sizeof(msg.body))
    {
        LOG_OUTPUT("Error:T = %s, len = %d,\n",msg.typestr(),len);
        return OS_NotEnoughMem;
    }

    if(len > 0)
    {
       // msg.body = pbody;
        memcpy(msg.body,pbody,len);
    }
    */

   //return SendToClient((char*)&msg,msg.len);
    return OS_NoErr;
}

OS_Error StreamServer::SendToClient(char* pbody, int len)
{
    
    for(OSQueueIter iter(&m_WorkingQueue);!iter.IsDone();)
    {
        OSQueueElem* pElem = iter.GetCurrent();
        iter.Next();
        if(NULL == pElem)
            continue;
        tClientInfo* pClientInfo = (tClientInfo*)pElem->GetEnclosingObject();
        if(NULL == pClientInfo)
            continue;
        //OSMutexLocker locker(&m_mutex);
        if( -1 == pClientInfo->pClient->AddToWriteBuffer(pbody,len))
            return OS_NotEnoughMem;
    }
    
    return OS_NoErr;
}

OS_Error StreamServer::SendStreamToClient(char* pbody, int len)
{/*
    for(OSQueueIter iter(&m_WorkingQueue);!iter.IsDone();)
    {
        OSQueueElem* pElem = iter.GetCurrent();
        iter.Next();
        if(NULL == pElem)
            continue;
        tClientInfo* pClientInfo = (tClientInfo*)pElem->GetEnclosingObject();
        if(NULL == pClientInfo)
            continue;

        //PackageHttp(pClientInfo,msg,len);
        cfg_msg cfgmsg;
        cfgmsg.type = M_GetJPEG;
        cfgmsg.len = len + MSG_HDR_LEN;
        memcpy(cfgmsg.body,pbody,len);
       // cfgmsg.body = pbody;
        //OSMutexLocker locker(&m_mutex);
        // 必须确保这两包数据一起发
        pClientInfo->pClient->AddToWriteBuffer((char*)&cfgmsg,cfgmsg.len);
        //pClientInfo->pClient->AddToWriteBufferRef((char*)&cfgmsg,cfgmsg.len,NULL,NULL);
        //pClientInfo->pClient->AddToWriteBuffer(pbody,len);
      //  char size[4] = {0};
     //   char *p = size;
        //char *pSize = (char*)malloc(4);
        //memset(pSize,0,4);
      //  SetUInt32(p,(UInt32)len);

        //for(int i = 0 ; i < 4 ; i++)
        //    printf("size[%d]=%d ",i,size[i]);
        //printf("\n");

     //   pClientInfo->pClient->AddToWriteBuffer(size,4);
     //   pClientInfo->pClient->AddToWriteBuffer(pbody,len);
    }*/
    return OS_NoErr;
}

int StreamServer::PackageHttp(ClientInfo* pClientInfo,char* msg,int len)
{
    char buffer[1024]={0};
    sprintf(buffer,"HTTP/1.0 200 OK\r\n" \
            "Access-Control-Allow-Origin: *\r\n" \
            STD_HEADER \
            "Content-Type: multipart/x-mixed-replace;boundary=" BOUNDARY "\r\n" \
            "\r\n" \
            "--" BOUNDARY "\r\n");
    pClientInfo->pClient->AddToWriteBuffer(buffer,strlen(buffer));
    pClientInfo->pClient->AddToWriteBuffer(msg,len);
    sprintf(buffer,"\r\n--" BOUNDARY "\r\n");
    pClientInfo->pClient->AddToWriteBuffer(buffer,strlen(buffer));
    return 0;
}
